package com.xiaoa.gossip.cluster.observer.impl;

import com.xiaoa.gossip.cluster.command.NodeState;
import com.xiaoa.gossip.cluster.command.PingCommand;
import com.xiaoa.gossip.cluster.command.PingNodeState;
import com.xiaoa.gossip.cluster.observer.NodeStateObserver;
import com.xiaoa.gossip.cluster.service.ClusterClientService;
import com.xiaoa.gossip.cluster.service.ClusterMsgFactory;
import com.xiaoa.gossip.cluster.service.ClusterService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

import static com.xiaoa.gossip.cluster.command.NodeState.NODE_STATE_P_FAIL;
import static com.xiaoa.gossip.cluster.command.NodeState.NODE_STATE_RUN;
import static com.xiaoa.gossip.cluster.constant.ClusterMethodConstant.FAIL_METHOD_NAME;

/**
 * @author luokaichuang
 * @time 2020/6/3
 */
@Component
public class NodePFailStateObserver implements NodeStateObserver {

    // pFail节点map key 为 pfile节点  value 为确认pfile节点数
    private Map<String, Set<String>> pFailMap = new ConcurrentHashMap<>();

    @Autowired
    private ClusterService clusterService;

    @Autowired
    private ClusterMsgFactory clusterMsgFactory;

    @Autowired
    private ClusterClientService clientService;

    @Override
    public int observerStats() {
        return NODE_STATE_P_FAIL;
    }

    @Override
    public void doProcessor(PingNodeState oldNode, PingNodeState newNode, NodeState pingNode) {

        // 被改变节点id
        String nodeId = newNode.getId();

        synchronized (nodeId) {
            // 通知的idList
            Set<String> pingNodeIdSet = pFailMap.get(nodeId);
            if (pingNodeIdSet == null) {
                pingNodeIdSet = new HashSet<>();
                pFailMap.put(nodeId, pingNodeIdSet);
            }
            pingNodeIdSet.add(pingNode.getId());

            // 获取正在运行状态的节点
            List<PingNodeState> runNodeList =  clusterService.noteList().stream()
                    .filter(n -> n.getState() == NODE_STATE_RUN).collect(Collectors.toList());
            // 统计当前存活的id list
            List<String> runNoteIdList = runNodeList.stream().map(n->n.getId())
                    .collect(Collectors.toList());

            // 如果当前运行所有节点都认为改节点已经fail ,向集群广播该节点 fail信息
            if(pingNodeIdSet.containsAll(runNoteIdList)){

                // 创建fail消息
                PingCommand failCommand =  clusterMsgFactory.createNodeFail(newNode);

                // 更新本机状态
                clusterService.doPongHandle(failCommand);

                // 广播fail信息
                clientService.doBroadcast(FAIL_METHOD_NAME ,runNodeList,failCommand );

            }

        }

    }


}
